package com.siteseat.crudeq.client.v1.request;

import java.io.EOFException;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;

import org.apache.log4j.Logger;

import com.siteseat.common.exception.InternalException;
import com.siteseat.common.misc.IOHelper;
import com.siteseat.common.security.CryptoHelper;
import com.siteseat.common.security.SecurityException;
import com.siteseat.crudeq.client.ClientMessage;
import com.siteseat.crudeq.client.ClientProfile;
import com.siteseat.crudeq.client.Connection;
import com.siteseat.crudeq.client.ErrorResponseException;
import com.siteseat.crudeq.client.InvalidClientMessage;
import com.siteseat.crudeq.client.InvalidResponseException;
import com.siteseat.crudeq.client.PublishResult;
import com.siteseat.crudeq.common.raw.MessageHelper;
import com.siteseat.crudeq.common.raw.protocol.v1.CodecHelper;
import com.siteseat.crudeq.common.raw.protocol.v1.ErrorMessage;
import com.siteseat.crudeq.common.raw.protocol.v1.V1RequestId;
import com.siteseat.crudeq.common.raw.protocol.v1.V1ResponseId;

/**
 *
 */
public class V1Connection implements Connection {
  
  private Logger       log = Logger.getLogger( V1Connection.class );

  private Socket       socket;

  private OutputStream upStream;

  private InputStream  downStream;

  private boolean      logined;
  
  /**
   * 
   *
   *
   */
  public V1Connection() {}
  
  /**
   * 
   *
   *
   * @param profile
   * @throws IOException 
   */
  public V1Connection( ClientProfile profile ) throws IOException {
    InetSocketAddress addr = new InetSocketAddress( profile.getServerHost(), profile.getServerPort() );
    connect( addr, profile.getConnectTimeoutByMilliseconds() );
    login( profile.getUserName(), profile.getPassword(), profile.getClientId(), profile.getLanguage() );
  }
  
  
  
  /**
   * Gets the upStream
   *
   * @return the upStream
   */
  protected OutputStream getUpStream() {
    return upStream;
  }

  /**
   * Gets the downStream
   *
   * @return the downStream
   */
  protected InputStream getDownStream() {
    return downStream;
  }

  /**
   * 
   */
  public void ensureLogined() {
    if( isLogined() == false ) {
      throw new InternalException( "Not logined yet" );
    }
  }


  /**
   * Gets the version
   * 
   * @return the version
   */
  public short getVersion() {
    return 1;
  }

  /**
   * 
   *
   * @return
   * @throws IOException
   * @throws SecurityException
   */
  public static byte[] encryptPassword( String userName, String password, String clientId ) throws IOException, SecurityException {
    long salt = System.currentTimeMillis();
    String encrypted = CryptoHelper.crudeEncrypt( userName, password, clientId, salt );
    return encrypted.getBytes( IOHelper.CHARSET );
  }
  
  /**
   * {@inheritDoc}
   */
  @Override
  public void login( String userName, String password, String clientId, String language ) throws IOException {
    if( isLogined() ) {
      return;
    }

    byte[] userNameBytes = userName.getBytes(IOHelper.CHARSET);
    
    byte[] passwordBytes;
    try {
      passwordBytes = encryptPassword( userName, password, clientId );
    } catch( SecurityException e ) {
      throw new IOException(e);
    }
    
    byte[] clientIdBytes = clientId.getBytes(IOHelper.CHARSET);
    byte[] languageBytes = language.getBytes(IOHelper.CHARSET);

    this.upStream.write( V1RequestId.LOGIN );

    int bodySize = 2 + 4 + userNameBytes.length + 4 + passwordBytes.length + 4 + clientIdBytes.length + 4 + languageBytes.length;
    IOHelper.writeInt( this.upStream, bodySize );

    IOHelper.writeShort( this.upStream, getVersion() );

    IOHelper.writeInt( this.upStream, userNameBytes.length );
    this.upStream.write( userNameBytes );

    IOHelper.writeInt( this.upStream, passwordBytes.length );
    this.upStream.write( passwordBytes );

    IOHelper.writeInt( this.upStream, clientIdBytes.length );
    this.upStream.write( clientIdBytes );

    IOHelper.writeInt( this.upStream, languageBytes.length );
    this.upStream.write( languageBytes );

    handleResponse( V1ResponseId.LOGINED );
  }

  /**
   * 
   *
   * @param address
   * @param timeoutByMilliseconds
   * @throws IOException
   */
  public void connect( InetSocketAddress address, int timeoutByMilliseconds ) throws IOException {
    if( isConnected() ) {
      return;
    }

    this.socket = newSocket();
    this.socket.connect( address, timeoutByMilliseconds );

    this.upStream = socket.getOutputStream();
    this.downStream = socket.getInputStream();
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public void disconnect() {
    this.logined = false;

    if( this.upStream != null ) {
      try {
        this.upStream.write( V1RequestId.DISCONNECT );
        IOHelper.writeInt( this.upStream, 0 );
      } catch( IOException e1 ) {
        this.upStream = null;
        this.downStream = null;
      }
    }
    
    if( this.socket != null ) {
      try {
        this.socket.close();
      } catch( IOException e ) {
        log.error( "Error during closing socket", e );
      }
      this.socket = null;
    }
  }

  /**
   * @return
   */
  protected Socket newSocket() {
    return new Socket();
  }

  /**
   * @param bodySize
   * @throws IOException
   */
  protected List<ClientMessage> handleMessageSetResponse( int bodySize ) throws IOException {
    boolean trace = log.isTraceEnabled();

    int queueCount = IOHelper.readInt( this.downStream );// queue count
    if( queueCount < 0 ) {
      throw new InvalidResponseException( "Corrupted queue count: " + queueCount );
    }
    if( trace ) {
      log.trace( "Queue count: " + queueCount );
    }

    List<ClientMessage> result = new ArrayList<ClientMessage>();
    
    for( int i = 0; i < queueCount; i++ ) {
      if( trace ) {
        log.trace( "Queue #" + i );
      }

      Integer queueId = Integer.valueOf( IOHelper.readInt( this.downStream ) );// //queue id
      if( trace ) {
        log.trace( "Queue id: " + queueId );
      }

      int chunkCount = IOHelper.readInt( this.downStream );// chunk count
      if( chunkCount <= 0 ) {
        throw new InvalidResponseException( "Corrupted chunk count: " + chunkCount );
      }
      if( trace ) {
        log.trace( "Chunk count: " + chunkCount );
      }

      if( bodySize <= 0 ) {
        throw new InvalidResponseException( "Corrupted total response size: " + bodySize );
      }

      for( int k = 0; k < chunkCount; k++ ) {
        long chunkId = IOHelper.readLong( this.downStream );// chunk id
        if( trace ) {
          log.trace( "Chunk id: " + chunkId );
        }

        int payloadBase = IOHelper.readInt( this.downStream );// 
        if( trace ) {
          log.trace( "Payload base: " + payloadBase );
        }
        
        int chunkSize = IOHelper.readInt( this.downStream );// chunk size
        if( trace ) {
          log.trace( "Chunk size: " + chunkSize );
        }
        if( chunkSize < 0 ) {
          throw new InvalidResponseException( "Corrupted chunk size: " + chunkSize );
        }
        if( chunkSize == 0 ) {
          continue;
        }

        byte[] payload = new byte[chunkSize];
        IOHelper.readBytes( this.downStream, payload );

        int payloadLocalOffset = 0;
        int messageSize = 0;
        while( ( payloadLocalOffset += messageSize ) < chunkSize ) {          
          boolean isValid = MessageHelper.isValidMagic( payload, payloadLocalOffset );
          ClientMessage m = isValid ? new ClientMessage() : new InvalidClientMessage();
          
          m.setPayload( payloadLocalOffset, payload );
          messageSize = m.getPayloadSize();
          
          m.setQueueId( queueId );
          m.setChunkId( chunkId );
          m.setPayloadBase( payloadBase );
          
          result.add( m );
          if( trace ) {
            log.trace( m );
          }
          if( !isValid ) {
            break;
          }
        }
      }
    }
    
    return result;
  }
  

  /**
   * @param bodySize
   * @throws IOException
   */
  protected ClientMessage handleMessageResponse( int bodySize ) throws IOException {
    int payloadSize = bodySize - 4 - 8 - 4;
    if( payloadSize <= 0 ) {
      throw new InvalidResponseException( "Corrupted message size: " + payloadSize );
    }
    
    int queueId = IOHelper.readInt( this.downStream );
    long chunkId = IOHelper.readLong( this.downStream );
    int payloadBase = IOHelper.readInt( this.downStream );
    
    byte[] payload = new byte[payloadSize];
    IOHelper.readBytes( this.downStream, payload );

    if( MessageHelper.isValidMagic( payload, 0 ) == false ) {
      throw new InternalException( "Invalid message magic value at payload base: " + (payloadBase + 0) );
    }
    
    int realPayloadSize = MessageHelper.readSize( payload, 0 );
    if( payloadSize != realPayloadSize ) {
      throw new InvalidResponseException( "Corrupted payload size: " + payloadSize + "<->" + realPayloadSize );
    }
    
    ClientMessage msg = new ClientMessage();

    msg.setPayloadBase( payloadBase );
    msg.setChunkId( chunkId );
    msg.setPayload( 0, payload );
    msg.setQueueId( Integer.valueOf(queueId) );

    return msg;
  }

  /**
   * @param bodySize
   * @throws InvalidResponseException
   */
  @SuppressWarnings( "unchecked" )
  protected List<ClientMessage> handleNoMessageResponse( int bodySize ) throws InvalidResponseException {
    if( bodySize != 0 ) {
      throw new InvalidResponseException( "corrupted NO_MESSAGE response" );
    }
    if( log.isTraceEnabled() ) {
      log.trace( "NO_MESSAGE" );
    }
    return Collections.EMPTY_LIST;
  }

  /**
   * @param bodySize
   * @throws InvalidResponseException
   * @throws IOException
   */
  protected PublishResult handlePublishedResponse( int bodySize ) throws IOException {
    InputStream ins = this.downStream;

    if( bodySize != 8/* chunk id */+ 4/* offset */) {
      throw new InvalidResponseException( "corrupted PUBLISHED response" );
    }

    long chunkId = IOHelper.readLong( ins );
    int offset = IOHelper.readInt( ins );

    PublishResult r = new PublishResult( chunkId, offset );

    if( log.isTraceEnabled() ) {
      log.trace( r );
    }

    return r;
  }

  /**
   * @param bodySize
   * @throws IOException
   */
  protected ErrorMessage handleErrorResponse( int bodySize ) throws IOException {
    if( bodySize < 8 ) {
      throw new InvalidResponseException( "corrupted ERROR response" );
    }

    ErrorMessage r = CodecHelper.readErrorMessage( this.downStream );

    if( log.isTraceEnabled() ) {
      log.trace( r );
    }

    return r;
  }

  /**
   * @param bodySize
   * @throws InvalidResponseException
   */
  protected void handleLoginedResponse( int bodySize ) throws InvalidResponseException {
    if( bodySize != 0 ) {
      throw new InvalidResponseException( "corrupted LOGINED response" );
    }

    this.logined = true;

    if( log.isTraceEnabled() ) {
      log.trace( "Logined" );
    }
  }
  
  /**
   * @param bodySize
   * @throws InvalidResponseException
   */
  protected ClientMessage handleMessageStoppedResponse( int bodySize ) throws InvalidResponseException {
    if( bodySize != 0 ) {
      throw new InvalidResponseException( "corrupted MESSAGE_STOPPED response" );
    }
    return ClientMessage.STOPPED;
  }
  
  /**
   * @param bodySize
   * @throws InvalidResponseException
   */
  protected void handleOKResponse( int bodySize ) throws InvalidResponseException {
    if( bodySize != 0 ) {
      throw new InvalidResponseException( "corrupted LOGINED response" );
    }
  }

  /**
   * @param responseId
   * @param bodySize
   * @throws InvalidResponseException
   */
  protected void handleUnknownResponse( short responseId, int bodySize ) throws InvalidResponseException {
    String msg = "unknown response: id=" + responseId + ", bodySize=" + bodySize;
    if( log.isTraceEnabled() ) {
      log.trace( msg );
    }
    throw new InvalidResponseException( msg );
  }

  /**
   * @param responseId
   * @param bodySize
   * @throws InvalidResponseException
   */
  protected void handleUnexpectedResponse( short receivedResponseId, short expectedResponseIds[], int bodySize )
          throws InvalidResponseException {
    String msg = "Unexpected response: id=" + receivedResponseId + "(expected is"
            + Arrays.toString( expectedResponseIds ) + "), bodySize=" + bodySize;
    if( log.isTraceEnabled() ) {
      log.trace( msg );
    }
    throw new InvalidResponseException( msg );
  }

  /**
   * @throws IOException
   * @throws InvalidResponseException
   * @throws ErrorResponseException
   */
  protected Object handleResponse( short... expectedResponseIds ) throws IOException {
    this.upStream.flush();
    
    InputStream ins = this.downStream;

    // Read the header
    byte receivedResponseId = (byte)ins.read();
    if( receivedResponseId == -1 ) {
      throw new EOFException();
    }

    int bodySize = IOHelper.readInt( ins );

    if( log.isTraceEnabled() ) {
      log.trace( "Response Id: " + receivedResponseId + ", body size: " + bodySize );
    }

    for( short expectedResponseId: expectedResponseIds ) {
      if( receivedResponseId == V1ResponseId.ERROR ) {
        ErrorMessage error = handleErrorResponse( bodySize );
        throw new ErrorResponseException( error );
      }

      if( expectedResponseId == receivedResponseId ) {
        Object r = null;
        switch( receivedResponseId ) {
          case V1ResponseId.MESSAGE:
            r = handleMessageResponse( bodySize );
            break;
          case V1ResponseId.MESSAGE_SET:
            r = handleMessageSetResponse( bodySize );
            break;
          case V1ResponseId.NO_MESSAGE:
            r = handleNoMessageResponse( bodySize );
            break;
          case V1ResponseId.PUBLISHED:
            r = handlePublishedResponse( bodySize );
            break;
          case V1ResponseId.MESSAGE_STOPPED:
            r = handleMessageStoppedResponse( bodySize );
            break;
          case V1ResponseId.LOGINED:
            handleLoginedResponse( bodySize );
            break;
          case V1ResponseId.OK:
            handleOKResponse( bodySize );
            break;
          default:
            handleUnexpectedResponse( receivedResponseId, expectedResponseIds, bodySize );
            break;
        }
        return r;
      }
    }

    handleUnexpectedResponse( receivedResponseId, expectedResponseIds, bodySize );
    return null;
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public boolean isConnected() {
    return this.socket != null && this.socket.isConnected();
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public boolean isLogined() {
    return this.logined;
  }

  /**
   * {@inheritDoc}
   */
  @Override
  protected void finalize() throws Throwable {
    disconnect();
  }

}
